package coc.core;

import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import coc.core.filter.MemoryBloomFilter;
import coc.core.url.Url;

/**
 * 抽象爬虫父类。
 */
public abstract class Crawler extends ThreadPoolExecutor implements
		Callable<CrawlerResult> {
	/** 日志 */
	protected Logger log = LoggerFactory.getLogger(getClass());
	/** 爬虫配置 */
	protected CrawlerConfig config;
	/** 爬虫结果 */
	protected CrawlerResult result;
	/** 作业结果列表 */
	protected LinkedBlockingQueue<Future<SpiderResult>> futures = new LinkedBlockingQueue<Future<SpiderResult>>();

	/**
	 * 构造方法。
	 * 
	 * @param config
	 *            爬虫配置
	 */
	public Crawler(CrawlerConfig config) {
		super(config.getThreadCount(), config.getThreadCount(), 0,
				TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
		config.setBloomFilter(new MemoryBloomFilter(0.0001, 100000));
		this.config = config;

		result = new CrawlerResult();
		result.setTaskId(config.getTaskId());
		result.setTaskName(config.getTaskName());
	}

	/**
	 * 过滤待处理的URL列表。
	 * 
	 * @param taskId
	 *            任务ID
	 * @param suspendUrls
	 *            待处理的URL列表
	 * @return 返回过滤后的URL列表。
	 */
	protected abstract List<Url> filterUrls(String taskId, List<Url> suspendUrls);

	/**
	 * 保存采集结果。
	 * 
	 * @param spiderResult
	 *            采集结果
	 * @return 返回采集的有效数据条数。
	 */
	protected abstract Integer saveSpiderResult(SpiderResult spiderResult);

	/**
	 * 完成任务。
	 */
	protected abstract void complete();

	@Override
	public CrawlerResult call() throws Exception {
		try {
			init();
			while (config.hasSuspendUrls() || !futures.isEmpty()) {
				process();
				Thread.sleep(config.getInterval());
			}
			destroy();
			return result;
		} catch (Exception e) {
			log.error("执行爬虫任务时发生异常。", e);
			throw e;
		} finally {
			shutdownNow();
		}
	}

	/**
	 * 初始化并启动任务。
	 */
	private void init() {
		log.info("开始任务[" + config.getTaskId() + "][" + config.getTaskName()
				+ "]");
		result.setStartupDate(new Date());
		config.addSuspendUrl(config.getStartUrl());
	}

	/**
	 * 处理作业。
	 */
	private void process() {
		for (Future<SpiderResult> future : futures) {
			processSpiderResult(future);
		}
		futures.clear();

		List<Url> suspendUrls = config.getSuspendUrls();
		suspendUrls = filterUrls(config.getTaskId(), suspendUrls);
		submitJobs(suspendUrls);
	}

	/**
	 * 处理采集结果。
	 * 
	 * @param future
	 *            采集结果
	 */
	private void processSpiderResult(Future<SpiderResult> future) {
		try {
			SpiderResult spiderResult = future.get(config.getJobTimeout(),
					TimeUnit.SECONDS);
			if (spiderResult.hasValidData()) {
				result.addCollectRecordCount(saveSpiderResult(spiderResult));
				log.debug("采集数据成功：[" + spiderResult.getUrl() + "]["
						+ spiderResult.getData() + "]");
			}
			result.addCompletedJobCount();
			log.debug("完成作业： " + spiderResult);
		} catch (Exception e) {
			result.addExceptionJobCount();
			log.debug("采集数据时发生异常。", e);
		}
	}

	/**
	 * 提交作业到任务队列。
	 * 
	 * @param urls
	 *            待处理作业的URL列表
	 */
	private void submitJobs(List<Url> urls) {
		for (Url url : urls) {
			Spider jobCaller = new Spider(url, config);
			futures.add(submit(jobCaller));
		}
		log.debug("提交待执行作业" + urls.size() + "个。");
	}

	/**
	 * 结束及清理任务。
	 */
	private void destroy() {
		result.setJobCount(new Long(getTaskCount()).intValue());
		result.setCompleteDate(new Date());
		complete();
		log.info("完成任务" + result);
	}
}
